package com.demo.sink;

import com.demo.domain.RealtimeEventEntity;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.container.RedisClusterContainer;
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainer;
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandsContainerBuilder;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.io.IOException;

/**
 * @author hornsey
 * @date 2022/10/18
 */
public class RealtimeTagRedisSink<IN> extends RedisSink<IN> {
	private final RedisCommand redisCommand;
	private final FlinkJedisConfigBase flinkJedisConfigBase;
	private RealtimeTagRedisMapper<IN> redisSinkMapper;
	private RedisCommandsContainer redisCommandsContainer;

	public RealtimeTagRedisSink(FlinkJedisConfigBase flinkJedisConfigBase, RealtimeTagRedisMapper<IN> redisSinkMapper) {
		super(flinkJedisConfigBase, (RedisMapper<IN>) redisSinkMapper);
		this.flinkJedisConfigBase = flinkJedisConfigBase;
		this.redisSinkMapper = redisSinkMapper;
		RedisCommandDescription redisCommandDescription = redisSinkMapper.getCommandDescription();
		this.redisCommand = redisCommandDescription.getCommand();
	}

	@Override
	public void invoke(IN input) throws Exception {
		String key = this.redisSinkMapper.getKeyFromData((RealtimeEventEntity) input);
		String value = this.redisSinkMapper.getValueFromData((RealtimeEventEntity) input);
		String additionalKey = this.redisSinkMapper.getAdditionalKey((RealtimeEventEntity) input);
		switch(this.redisCommand) {
			case RPUSH:
				this.redisCommandsContainer.rpush(key, value);
				break;
			case LPUSH:
				this.redisCommandsContainer.lpush(key, value);
				break;
			case SADD:
				this.redisCommandsContainer.sadd(key, value);
				break;
			case SET:
				this.redisCommandsContainer.set(key, value);
				break;
			case PFADD:
				this.redisCommandsContainer.pfadd(key, value);
				break;
			case PUBLISH:
				this.redisCommandsContainer.publish(key, value);
				break;
			case ZADD:
				this.redisCommandsContainer.zadd(additionalKey, value, key);
				break;
			case HSET:
				this.redisCommandsContainer.hset(additionalKey, key, value);
				break;
			default:
				throw new IllegalArgumentException("Cannot process such data type: " + this.redisCommand);
		}
	}


	@Override
	public void open(Configuration parameters) throws Exception {
		this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
	}

	@Override
	public void close() throws IOException {
		if (this.redisCommandsContainer != null) {
			this.redisCommandsContainer.close();
		}

	}
}
